package org.nbict.iot.trident.influx.mapper;

import org.apache.storm.tuple.ITuple;
import org.influxdb.dto.Point;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Created by songseven on 18/7/3.
 */
public class InfluxdbMapper implements Serializable {

    private String measurement;
    private List<String> tags;
    private List<String> fields;
    private String timestamp;

    public InfluxdbMapper(String measurement, List tags, List fields, String timestamp) {

        this.measurement = measurement;
        this.tags = tags;
        this.fields = fields;
        this.timestamp = timestamp;
    }

    public Point toPoint(ITuple tuple) {

        Map tagMap = new HashMap<String, String>();
        tags.forEach(t->{
            tagMap.put(t, tuple.getStringByField(t));
        });

        Map fieldMap = new HashMap<String, Number>();
        fields.forEach(f->{
            fieldMap.put(f, tuple.getValueByField(f));
        });

        Point point = Point.measurement(measurement)
                .tag(tagMap) //tag
                .time(tuple.getLongByField(timestamp), TimeUnit.SECONDS) //时间戳
                .fields(fieldMap) //field
                .build();

        return point;
    }
}
